package com.zangs.emqx_expand.emqx.service;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import com.zangs.emqx_expand.common.constant.MqttConstants;
import com.zangs.emqx_expand.common.utils.uuid.IdUtils;
import com.zangs.emqx_expand.emqx.domain.MqttUserOnline;
import com.zangs.emqx_expand.emqx.mqtt.MqttConfig;
import org.apache.commons.lang3.ObjectUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

@Component
public class MqttClientService {

    private MqttClient mqttClient;
    @Resource
    private MqttConfig mqttConfig;


    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    private volatile boolean isConnected = false; // 用于跟踪是否已连接


    /**
     * $SYS/brokers/{node}/clients/{clientid}/connected 系统主题，上线事件。当任意客户端上线时，EMQX 就会发布该主题的消息
     * ^\\$SYS/brokers/([^/]+)/clients/([^/]+)/connected$
     * ^ 表示字符串的开始。
     * \\$SYS/brokers/ 匹配固定的 $SYS/brokers/ 部分。
     * ([^/]+) 匹配 {node} 部分，[^/]+ 表示一个或多个非 / 的字符。
     * /clients/ 匹配固定的 /clients/ 部分。
     * ([^/]+) 匹配 {clientid} 部分。
     * /connected$ 匹配固定的 /connected 部分，$ 表示字符串的结束。
     */
    private static final String TOPIC_PATTERN_CONNECTED = "^\\$SYS/brokers/([^/]+)/clients/([^/]+)/connected$";
    //离线事件
    private static final String TOPIC_PATTERN_DISCONNECTED = "^\\$SYS/brokers/([^/]+)/clients/([^/]+)/disconnected$";



    @PostConstruct
    public void init() {
        mqttConfig.setClientId(mqttConfig.getClientId() + IdUtils.simpleUUID());
        scheduler.scheduleAtFixedRate(this::tryConnect, 0, 5, TimeUnit.SECONDS);
    }

    private void tryConnect() {
        try {
            if (!isConnected) {
                MqttClient client = new MqttClient(mqttConfig.getBrokerUrl(), mqttConfig.getClientId(), new MemoryPersistence());
                MqttConnectOptions options = new MqttConnectOptions();
                options.setUserName(mqttConfig.getUsername());
                options.setPassword(mqttConfig.getPassword().toCharArray());
                options.setCleanSession(mqttConfig.isCleanSession());
                options.setAutomaticReconnect(mqttConfig.isAutoReconnect());

                client.connect(options);

                for (String topic : mqttConfig.getTopics()) {
                    client.subscribe(topic);
                }

                this.mqttClient = client;
                this.mqttClient.setCallback(new MqttCallback() {
                    @Override
                    public void connectionLost(Throwable throwable) {
                        // 连接丢失后，可以重新设置isConnected为false，允许再次尝试连接
                        isConnected = false;
                    }

                    @Override
                    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {

                        // 处理消息
                        String msg = new String(mqttMessage.getPayload(), MqttConstants.MSG_CODE);
                        if (Pattern.compile(TOPIC_PATTERN_CONNECTED).matcher(topic).matches() || Pattern.compile(TOPIC_PATTERN_DISCONNECTED).matcher(topic).matches()) {
                            //其他值参考emqx文档
                            Map<String, Object> map = JSON.parseObject(msg, new TypeReference<Map<String, Object>>() {});
                            String clientid = (String) map.get("clientid");
                            Long connected_at = (Long) map.get("connected_at");
                            MqttUserOnline query = new MqttUserOnline();
                            query.setClientId(clientid);
                            MqttUserOnline mqttUserOnline = iMqttUserOnlineService.selectMqttUserOnline(query);
                            mqttUserOnline.setOnline(ObjectUtils.isNotEmpty(connected_at));
                            iMqttUserOnlineService.updateMqttUserOnline(mqttUserOnline);
                        }

                    }

                    @Override
                    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                        // 消息发送完成处理
                    }
                });

                // 连接成功后取消定时任务
                isConnected = true;
                scheduler.shutdown();
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    @Resource
    private IMqttUserOnlineService iMqttUserOnlineService;


    @PreDestroy
    public void destroy() {
        scheduler.shutdownNow();
    }
}
